package org.raftkv.raft.core.rawnode;

import org.raftkv.commons.function.TripleConsumer;
import org.raftkv.raft.core.Config;
import org.raftkv.raft.core.Raft;
import org.raftkv.raft.core.node.Peer;
import org.raftkv.raft.core.node.Ready;
import org.raftkv.raft.core.node.SoftState;
import org.raftkv.raft.core.node.enums.SnapshotStatus;
import org.raftkv.raft.core.rawnode.enums.ProgressType;
import org.raftkv.raft.core.rawnode.exceptions.StepLocalMsgException;
import org.raftkv.raft.core.rawnode.exceptions.StepPeerNotFoundException;
import org.raftkv.raft.core.status.BasicStatus;
import org.raftkv.raft.core.status.Status;
import org.raftkv.raft.exceptions.RaftException;
import org.raftkv.raft.pb.ConfState;
import org.raftkv.raft.pb.Entry;
import org.raftkv.raft.pb.HardState;
import org.raftkv.raft.pb.IConfChange;
import org.raftkv.raft.pb.Message;
import org.raftkv.raft.pb.enums.ConfChangeType;
import org.raftkv.raft.pb.enums.EntryType;
import org.raftkv.raft.pb.enums.MessageType;
import org.raftkv.raft.pb.impl.ConfChange;
import org.raftkv.raft.tracker.Progress;
import org.raftkv.raft.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;


/**
 * Raft 节点默认实现
 *
 * @author wzy
 */
public class RawNode {

    private static final Logger LOGGER = LoggerFactory.getLogger(RawNode.class);

    private Raft raft;

    private SoftState prevSoftSt;

    private HardState prevHardSt;

    private RawNode() {

    }

    public static RawNode newRawNode(Config config) {
        Raft r = Raft.newRaft(config);
        RawNode node = new RawNode();
        node.raft = r;
        node.prevHardSt = r.hardState();
        node.prevSoftSt = r.softState();
        return node;
    }

    public void tick() {
        this.raft.getTick().tick();
    }

    public void tickQuiesced() {
        this.raft.setElectionElapsed(this.raft.getElectionElapsed() + 1);
    }

    public void campaign() {
        this.raft.step(Message.builder()
                .type(MessageType.MsgHup)
                .build());
    }

    public void propose(byte[] data) {
        ArrayList<Entry> ents = new ArrayList<>();
        Entry e = new Entry();
        e.setData(data);
        this.raft.step(Message.builder()
                .type(MessageType.MsgProp)
                .from(this.raft.getId())
                .entries(ents)
                .build());
    }

    public void proposeConfChange(IConfChange cc) {
        Message m = IConfChange.confChangeToMsg(cc);
        this.raft.step(m);
    }

    public ConfState applyConfChange(IConfChange cc) {
        return this.raft.applyConfChange(cc.asV2());
    }

    public void step(Message m) {
        if (Utils.isLocalMsg(m.getType())) {
            throw new StepLocalMsgException();
        }
        Progress pr = this.raft.getPrs().getProgress().getProgressMap().get(m.getFrom());
        if (pr != null || !Utils.isResponseMsg(m.getType())) {
            this.raft.step(m);
        }

        throw new StepPeerNotFoundException();
    }

    public Ready ready() {
        Ready rd = this.readyWithoutAccept();
        acceptReady(rd);
        return rd;
    }

    public Ready readyWithoutAccept() {
        return Ready.newReady(this.raft, this.prevSoftSt, this.prevHardSt);
    }

    public void acceptReady(Ready rd) {
        if (rd.getSoftState() != null) {
            this.prevSoftSt = rd.getSoftState();
        }

        if (rd.getReadStates() != null && rd.getReadStates().size() != 0) {
            this.raft.setReadStates(null);
        }
        this.raft.setMsgs(null);
    }

    public boolean hasReady() {
        Raft r = this.raft;
        if (!r.softState().equals(this.prevSoftSt)) {
            return true;
        }

        HardState hardSt = r.hardState();
        if (!HardState.isEmptyHardState(hardSt) && !HardState.isHardStateEqual(hardSt, this.prevHardSt)) {
            return true;
        }

        if (r.getRaftLog().hasPendingSnapshot()) {
            return true;
        }

        if ((r.getMsgs() != null && r.getMsgs().size() > 0) || (r.getRaftLog().unstableEntries() != null && r.getRaftLog().unstableEntries().size() > 0)
                || r.getRaftLog().hasNextEnts()) {
            return true;
        }

        if (r.getReadStates().size() != 0) {
            return true;
        }

        return false;
    }

    public void advance(Ready rd) {
        if (!HardState.isEmptyHardState(rd.getHardState())) {
            this.prevHardSt = rd.getHardState();
        }
        this.raft.advance(rd);
    }

    public Status status() {
        return Status.getStatus(this.raft);
    }

    public BasicStatus basicStatus() {
        return Status.getBasicStatus(this.raft);
    }

    public void withProgress(TripleConsumer<Long, ProgressType, Progress> visitor) {
        this.raft.getPrs().visit((id, prs) -> {
            ProgressType typ = ProgressType.ProgressTypePeer;
            if (prs.getProgressMap().get(id).isLearner()) {
                typ = ProgressType.ProgressTypeLearner;
            }
            Progress p = prs.getProgressMap().get(id).clone();
            p.setInflights(null);
            visitor.accept(id, typ, p);
        });
    }

    public void reportUnreachable(long id) {
        this.raft.step(new Message.MessageBuilder()
                .type(MessageType.MsgUnreachable)
                .from(id)
                .build());
    }

    public void reportSnapshot(long id, SnapshotStatus status) {
        boolean rej = status == SnapshotStatus.SnapshotFailure;

        this.raft.step(new Message.MessageBuilder()
                .type(MessageType.MsgSnapStatus)
                .from(id)
                .reject(rej)
                .build());
    }

    public void transferLeader(long transferee) {
        this.raft.step(new Message.MessageBuilder()
                .type(MessageType.MsgTransferLeader)
                .from(transferee)
                .build());
    }

    public void readIndex(byte[] rctx) {
        List<Entry> ents = new ArrayList<>();
        Entry e = new Entry();
        e.setData(rctx);
        ents.add(e);

        this.raft.step(new Message.MessageBuilder()
                .type(MessageType.MsgReadIndex)
                .entries(ents)
                .build());
    }

    public void bootstrap(List<Peer> peers) {
        if (peers.size() == 0) {
            throw new RaftException("must provide at least one peer to Bootstrap");
        }

        long lastIndex = this.raft.getRaftLog().getStorage().lastIndex();
        if (lastIndex != 0) {
            throw new RaftException("can't bootstrap a nonempty Storage");
        }

        this.prevHardSt = HardState.EMPTY;
        this.raft.becomeFollower(1, Raft.NONE);
        List<Entry> ents = new ArrayList<>(peers.size());
        for (int i = 0; i < peers.size(); i++) {
            Peer peer = peers.get(i);

            ConfChange cc = new ConfChange();
            cc.setType(ConfChangeType.ConfChangeAddNode);
            cc.setNodeId(peer.getId());
            cc.setContext(peer.getContext());

            byte[] data = cc.marshal();

            Entry e = new Entry();
            e.setType(EntryType.EntryConfChange);
            e.setTerm(1);
            e.setIndex(i + 1);
            e.setData(data);
            ents.set(i, e);
        }
        raft.getRaftLog().append(ents);

        raft.getRaftLog().setCommitted(ents.size());
        for (Peer peer : peers) {
            ConfChange cc = new ConfChange();
            cc.setNodeId(peer.getId());
            cc.setType(ConfChangeType.ConfChangeAddNode);
            raft.applyConfChange(cc.asV2());
        }
    }

    public Raft getRaft() {
        return raft;
    }

    public void setRaft(Raft raft) {
        this.raft = raft;
    }

    public HardState getPrevHardSt() {
        return prevHardSt;
    }

    public void setPrevHardSt(HardState prevHardSt) {
        this.prevHardSt = prevHardSt;
    }

    public SoftState getPrevSoftSt() {
        return prevSoftSt;
    }

    public void setPrevSoftSt(SoftState prevSoftSt) {
        this.prevSoftSt = prevSoftSt;
    }

}
